package com.yh.csx.bsf.ods;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.Message;
import com.yh.csx.bsf.core.util.LogUtils;

import lombok.Data;
import lombok.var;

import java.util.ArrayList;
import java.util.List;

public class BaseCanalConsumer extends AbstractCanalConsumer {

    protected List<Operate> batchOperate = new ArrayList<>();
    public static enum OperateEnum{
        insert,
        update,
        delete,
        alterTable,
    }

    @Data
    public static class Operate{
        OperateEnum operateEnum;
        CanalEntry.Header header;
        List<CanalEntry.Column> beforeColumns;
        List<CanalEntry.Column> afterColumns;
    }


    /**
     * 强烈建议捕获异常
     * @param header
     * @param afterColumns
     */
    @Override
    public void insert(CanalEntry.Header header, List<CanalEntry.Column> afterColumns) throws Exception {
        Operate o = new Operate();
        o.header=header;o.afterColumns=afterColumns;o.beforeColumns=null;o.operateEnum= OperateEnum.insert;
        batchOperate.add(o);
    }



    /**
     * 强烈建议捕获异常
     * @param header
     * @param beforeColumns 变化之前的列数据
     * @param afterColumns 变化之后的列数据
     */
    @Override
    public void update(CanalEntry.Header header,List<CanalEntry.Column> beforeColumns,List<CanalEntry.Column> afterColumns)throws Exception {
        Operate o = new Operate();
        o.header=header;
        o.afterColumns=afterColumns;
        o.beforeColumns=beforeColumns;
        o.operateEnum= OperateEnum.update;
        batchOperate.add(o);

    }

    /**
     * 强烈建议捕获异常
     * @param header
     * @param beforeColumns 删除之前的列数据
     */
    @Override
    public void delete(CanalEntry.Header header,List<CanalEntry.Column> beforeColumns)throws Exception {
        Operate o = new Operate();
        o.header=header;
        o.afterColumns=null;
        o.beforeColumns=beforeColumns;
        o.operateEnum= OperateEnum.delete;
        batchOperate.add(o);
    }

    @Override
    public void connectorBatchArk(long batchId)  throws Exception {
        for(var o :batchOperate)
        {
            if(o.operateEnum == OperateEnum.insert) {
                printColumn(o.operateEnum.name(), o.header.getTableName(), o.afterColumns);
            }else if(o.operateEnum == OperateEnum.update)
            {
                printColumn(o.operateEnum.name(), o.header.getTableName(), o.afterColumns);
            }
            else if(o.operateEnum == OperateEnum.delete)
            {
                printColumn(o.operateEnum.name(), o.header.getTableName(), o.getBeforeColumns());
            }
            else if(o.operateEnum == OperateEnum.alterTable)
            {
                printColumn(o.operateEnum.name(), o.header.getTableName(), o.getAfterColumns());
            }
        }

    }

    private static void printColumn(String action, String tableName,List<CanalEntry.Column> columns) {
    	LogUtils.debug(BaseCanalConsumer.class,"ods","表:"+ tableName+" 操作:"+action+" ");
        for (CanalEntry.Column column : columns) {
        	LogUtils.debug(BaseCanalConsumer.class,"ods",column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());          
        }
        LogUtils.debug(BaseCanalConsumer.class,"ods","*********************************************");
    }



	@Override
	public boolean insertBatch(Entry[] entries) throws Exception {
		// TODO Auto-generated method stub
		return true;
		
	}



	@Override
	public boolean updateBatch(Entry[] entries) throws Exception {
		// TODO Auto-generated method stub
		return true;
		
	}



	@Override
	public boolean deleteBatch(Entry[] entries) throws Exception {
		// TODO Auto-generated method stub
		return true;
		
	}



	@Override
	protected boolean processMessages(Message message) throws Exception {
		// TODO Auto-generated method stub
		return false;
	}
}
